package com.xcc.dao;


import com.google.common.collect.Lists;
import com.xcc.anno.ColumnRef;
import com.xcc.anno.RowkeyRef;
import com.xcc.anno.TableRef;
import com.xcc.constant.Names;
import org.apache.commons.configuration.ConfigurationFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import javax.naming.Name;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Scanner;


public abstract class BaseDao {

    private ThreadLocal<Connection> connectionThreadLocal = new ThreadLocal<Connection>();
    private ThreadLocal<Admin> adminThreadLocal = new ThreadLocal<Admin>();

    protected void createNamespace(String name) {
        try {
            Admin admin = getAdmin();
            admin.createNamespace(NamespaceDescriptor.create(name).build());
        } catch (NamespaceExistException e) {
            System.out.println("当前命名空间已经存在!");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取连接对象
     */
    protected synchronized Connection getConnection() throws Exception {
        Connection connection = connectionThreadLocal.get();
        if (connection == null) {
            connection = ConnectionFactory.createConnection(HBaseConfiguration.create());
            connectionThreadLocal.set(connection);
        }
        return connection;
    }

    /**
     * 获取admin对象
     */
    protected synchronized Admin getAdmin() throws Exception {
        Admin admin = adminThreadLocal.get();
        if (admin == null) {
            admin = getConnection().getAdmin();
            adminThreadLocal.set(admin);
        }
        return admin;
    }

    /**
     * 初始化
     */
    protected void start() throws Exception {
        getConnection();
        getAdmin();
    }

    /**
     * 关闭资源
     */
    protected void end() throws Exception {
        Admin admin = adminThreadLocal.get();
        if (admin != null) {
            admin.close();
            adminThreadLocal.remove();
        }
        Connection connection = connectionThreadLocal.get();
        if (connection != null) {
            connection.close();
            connectionThreadLocal.remove();
        }
    }

    /**
     * 创建表，如果存在就删除再创建
     */
    protected void createTableXX(String name, String... families) throws Exception {
        createTableXX(name, null, null, families);
    }

    protected void createTableXX(String name, Integer regionCount, String... families) throws Exception {
        createTableXX(name, null, regionCount, families);
    }

    protected void createTableXX(String name, String coprocessor, Integer regionCount, String... families) throws Exception {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        if (admin.tableExists(tableName)) {
            deleteTable(name);
        }
        HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
        if (families == null || families.length == 0) {
            hTableDescriptor.addFamily(new HColumnDescriptor(Names.FAMILY.getValue()));
        } else {
            for (String family : families) {
                hTableDescriptor.addFamily(new HColumnDescriptor(family));
            }
        }

        if (coprocessor != null && !"".equals(coprocessor)) {
            hTableDescriptor.addCoprocessor(coprocessor);
        }

        if (regionCount == null || regionCount <= 1) {
            admin.createTable(hTableDescriptor);
        } else {
            byte[][] rowKey = genRowKey(regionCount);
            admin.createTable(hTableDescriptor, rowKey);
        }
    }

    protected byte[][] genRowKey(Integer regionCount) {
        int count = regionCount - 1;
        byte[][] rowkey = new byte[count][];
        for (int i = 0; i < count; i++) {
            String strKey = i + "|";
            rowkey[i] = Bytes.toBytes(strKey);
        }
        return rowkey;
    }

    public static void main(String[] args) throws Exception {
//        BaseDao baseDao = new BaseDao();
//        baseDao.createTableXX(Names.TABLE.getValue(),4,"test","test2","yes");
//        baseDao.deleteTable(Names.TABLE.getValue());
//        System.out.println(getRowkeyRegion("17601009173", "20190101235959",4));
    }

    /**
     * 删除表
     *
     * @param name 表名
     */
    protected void deleteTable(String name) throws Exception {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }

    /**
     * 获取分区号
     * 根据手机号后4位随机性 与 拨号时间月份之前的数值进行异或运算 得出随机的分区号
     * 保证同一个手机号的同一个月在同一个分区内 提高查询效率
     *
     * @param tel      手机号:1331234 5678
     * @param callTime 拨号时间:201901 12231111
     */
    public int getRowkeyRegion(String tel, String callTime, Integer regionCount) {
        String last = tel.substring(tel.length() - 4);
        String first = callTime.substring(0, 6);
        int abs = Math.abs(Integer.valueOf(last) ^ Integer.valueOf(first));
        int region = abs % regionCount;
        return region;
    }

    /**
     * 添加数据
     */
    public void put(TableName tableName, Put put) throws Exception {
        Table table = getConnection().getTable(tableName);
        table.put(put);
        table.close();
    }

    /**
     * 扫描指定表的指定路径内容
     *
     * @param tableName 表名
     * @param startRow  开始
     * @param endRow    结束
     */
    public ResultScanner scan(String tableName, String startRow, String endRow) throws Exception {
        Table table = getConnection().getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(endRow));
        ResultScanner resultScanner = table.getScanner(scan);
        return resultScanner;
    }

    public void put(Object object) throws Exception {
        Class clazz = object.getClass();
        TableRef tableRef = (TableRef) clazz.getAnnotation(TableRef.class);
        String tableName = tableRef.value();
        if (tableName == null || "".equals(tableName)) {
            System.out.println("表名不存在!");
            return;
        }
        String rowkey = "";
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            RowkeyRef rowkeyRef = field.getAnnotation(RowkeyRef.class);
            if (rowkeyRef != null) {
                //私有方法放开
                field.setAccessible(true);
                rowkey = rowkeyRef.value();
                if (rowkey == null || "".equals(rowkey)) {
                    //获取属性值
                    rowkey = field.get(object).toString();
//                    System.out.println(rowkey);
                }
            }
        }
        Put put = new Put(Bytes.toBytes(rowkey));
        for (Field field : fields) {
            ColumnRef columnRef = field.getAnnotation(ColumnRef.class);
            if (columnRef != null) {
                //私有方法放开
                field.setAccessible(true);
                String column = columnRef.value();
                if (column == null || "".equals(column)) {
                    //获取属性名
                    String name = field.getName();
                    //获取属性值
                    String value = field.get(object).toString();
                    if (value != null && !"".equals(value)) {
                        put.addColumn(Bytes.toBytes(Names.CALLER.getValue()), Bytes.toBytes(name), Bytes.toBytes(value));
                    }
                }
            }
        }
        TableName tn = TableName.valueOf(tableName);
        put(tn, put);
    }
}
